package com.wish.group05.system.kafka;

import com.alibaba.fastjson.JSONObject;
import com.wish.group05.common.redis.service.RedisService;
import com.wish.group05.system.api.domain.SysOperLog;
import com.wish.group05.system.service.ISysOperLogService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumer0 implements AcknowledgingMessageListener<String,String> {
    private static final Logger log = LoggerFactory.getLogger(KafkaConsumer0.class);

    @Autowired
    private RedisService redisService;
    @Autowired
    private ISysOperLogService operLogService;
    @Override
    @KafkaListener(topicPartitions = @TopicPartition(topic = "test",partitions = {"0"}),groupId = "test-group")
    public void onMessage(ConsumerRecord<String, String> data, Acknowledgment acknowledgment) {
        String key = data.key();
        System.err.println("kafka开始消费");
        // 防止消息重复消费
        if(!redisService.setCacheLock(key)){
            log.info("kafka重复消费");
            return;
        }
        log.info("kafka开始消费："+key);
        String value = data.value();
        SysOperLog sysOperLog = JSONObject.parseObject(value, SysOperLog.class);
        // 保存日志到数据库
        int i = operLogService.insertOperlog(sysOperLog);
        if (i <= 0) {
            log.error("kafka:"+key+"消费失败");
            return;
        }
        // 删除redis中的备份
        redisService.deleteObject("message:"+key);
        // 提交偏移量
        acknowledgment.acknowledge();
    }
}
